package insight;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author huu
 * @create 2025-08-26-10:22
 * @Description: kafak缓冲池
 */
public class BufferPool {

    // 总长度
    private final int totalSize;

    // 槽位长度
    private final int slotSize;

    // 当前可用长度
    private int free;

    //双向队列，可复用槽位
    private final Deque<ByteBuffer> slotQueue = new ArrayDeque<>();

    //保证线程安全的锁
    private final Lock lock = new ReentrantLock();

    //保存condition，等待被唤醒
    private final Deque<Condition> waiters = new ArrayDeque<>();

    public BufferPool(int totalSize, int slotSize) {
        this.totalSize = totalSize;
        this.slotSize = slotSize;
        this.free = totalSize;
    }

    public ByteBuffer allocate(int size , long timeout) throws InterruptedException {
        //避免后面等待一直等待
        if(size > totalSize || size <= 0){
            throw new RuntimeException("size must be between 0 and " + totalSize);
        }
        lock.lock();
        try{
            //1.传入size刚好为一个槽位，同时也有剩余的槽位，则直接返回
            //slotQueue.size() > 0
            if(size == slotSize && !slotQueue.isEmpty()){
                return slotQueue.pollFirst();
            }
            //2.剩余的长度+剩余的槽位队列数量+一个槽位大小 >= size, 就说明有
            if((free + slotQueue.size() * slotSize) >= size){
                //释放
                freeUp(size);
                free -= size;
                //ByteBuffer.allocate创建一个具有指定容量的堆分配字节缓冲区
                return ByteBuffer.allocate(size);
            }// total->10   1->7  2->4
            //3.当前内存不满足我可申请内存，需要等待，创建一个与 Lock 关联的条件变量
            Condition condition = lock.newCondition();
            waiters.addLast(condition);
            //超时时间
            long remainTime = timeout;
            try {
                while (true) {
                    long start = System.currentTimeMillis();
                    boolean wakeup = condition.await(remainTime, TimeUnit.MILLISECONDS);
                    if(!wakeup){
                        throw new RuntimeException("超时");
                    }
                    if (size == slotSize && slotQueue.size() > 0) {
                        return slotQueue.pollFirst();
                    }
                    if ((free + slotQueue.size() * slotSize) >= size) {
                        freeUp(size);
                        free -= size;
                        return ByteBuffer.allocate(size);
                    }
                    remainTime -= System.currentTimeMillis() - start;
                }
            }finally{
                waiters.remove(condition);
            }

        }finally {
            if(!waiters.isEmpty() && !(free == 0 && slotQueue.isEmpty())){
                waiters.peek().signal();
            }
            lock.unlock();
        }


    }

    private void freeUp(int size) {
        //不断取消可复用槽，加入剩余内存
        while(free < size && !slotQueue.isEmpty()){
            free += slotQueue.pollFirst().capacity();
        }
    }

    public void deallocate(ByteBuffer byteBuffer) {
        lock.lock();
        try {
            if(byteBuffer.capacity() == this.slotSize){
                byteBuffer.clear();
                slotQueue.addLast(byteBuffer);
            }else{
                free += byteBuffer.capacity();
            }
            if(!waiters.isEmpty()){
                waiters.peek().signal();
            }
        }finally{
            lock.unlock();
        }
    }

}
